package com.fwmagic.flink.projectcase.queryactivitycase.async;

import com.alibaba.druid.pool.DruidDataSource;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.functions.async.ResultFuture;
import org.apache.flink.streaming.api.functions.async.RichAsyncFunction;

import java.sql.Connection;
import java.sql.PreparedStatement;
import java.sql.ResultSet;
import java.sql.SQLException;
import java.util.Collections;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.Future;
import java.util.function.Supplier;

public class AsyncQueryMysqlFunction extends RichAsyncFunction<String, String> {

    private transient DruidDataSource dataSource;

    private transient ExecutorService executorService;

    @Override
    public void open(Configuration parameters) throws Exception {
        executorService = Executors.newFixedThreadPool(20);
        dataSource = new DruidDataSource();
        dataSource.setDriverClassName("com.mysql.jdbc.Driver");
        dataSource.setUsername("root");
        dataSource.setPassword("123456");
        dataSource.setUrl("jdbc:mysql://localhost:3306/flink-test?useUnicode=true&characterEncoding=UTF-8");
        dataSource.setInitialSize(5);
        dataSource.setMinIdle(10);
        dataSource.setMaxActive(20);
    }

    @Override
    public void asyncInvoke(String id, ResultFuture<String> resultFuture) throws Exception {
        Future<String> future = executorService.submit(() -> {
            return queryFromMysql(id);
        });

        CompletableFuture.supplyAsync(new Supplier<String>() {
            @Override
            public String get() {
                try {
                    return future.get();
                } catch (Exception e) {
                  return null;
                }
            }
        }).thenAccept((String dbResult) ->{
            resultFuture.complete(Collections.singleton(dbResult));
        });
    }

    private String queryFromMysql(String id) throws Exception{
        String sql = "select name from t_activities where id =?";
        Connection connection = null;
        PreparedStatement ps = null;
        String result = null;
        ResultSet rs = null;
        try {
            connection = dataSource.getConnection();
            ps = connection.prepareStatement(sql);
            ps.setString(1, id);
            rs = ps.executeQuery();
            while (rs.next()){
                result = rs.getString(1);
            }
        }finally {
            if(rs!=null) rs.close();
            if(ps!=null) ps.close();
            if(connection!=null) connection.close();
        }

        if(result!=null){
            //放入缓存中
        }
        return result;
    }

    @Override
    public void close() throws Exception {
        dataSource.close();
        executorService.shutdown();
    }
}
